package zx.lrn.notes.multi_threading;

import java.nio.file.Paths;
import java.util.concurrent.*;
import java.util.stream.Stream;

/**
 *
 * @ClassName: CompletableFutureLrn
 * @Description: CompletableFuture学习 CompletableFuture可以实现无等待的异步计算。
 *               我个人的理解：CompletableFuture内包含的数据驱动着每一步的计算，每一步的完成，会自动进行下一步的计算（如果有），这一切都是异步的
 * @Author: zhaoxuan
 * @Date: 2020/3/13 0:19
 **/
public class CompletableFutureLrn {
    private static ExecutorService pool = Executors.newCachedThreadPool();

    // CompletableFuture的承诺赋值
    private static void promiseAssign() throws ExecutionException, InterruptedException {
        CompletableFuture<String> willPromise = new CompletableFuture<>();
        pool.submit(() -> {
            sleepNoExp(3);
            willPromise.complete("ok"); // 5S后显式赋值
        });
        System.out.println(willPromise.get());
    }

    // CompletableFuture的组合使用，变相的事件驱动，促使CompletableFuture向前流动
    private static void combination() throws ExecutionException, InterruptedException {
        // 模拟从硬盘中读取数据并乘2处理，耗时3秒
        CompletableFuture<Integer> fromIO = CompletableFuture.supplyAsync(() -> {
            sleepNoExp(3);
            System.out.println(Thread.currentThread().getName() + ":io ok");
            return 10;
        }, pool).thenApply(n -> n << 1);

        // 期间进行别的操作
        System.out.println("main sing");

        // 模拟从网络中获取数据，获取后做处理，最后和IO中的数据做乘法
        int result = CompletableFuture.supplyAsync(() -> "1,2,3,4,5", pool)
            .thenComposeAsync(str -> CompletableFuture.supplyAsync(() -> {
                int netResult = CompletableFutureLrn.sumOfStr(str);
                System.out.println(Thread.currentThread().getName() + ":net ok");
                return netResult;
            }, pool), pool).thenCombineAsync(fromIO, (net, io) -> net * io, pool).get();
        System.out.println(result);
    }

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        promiseAssign();
        combination();
        pool.shutdown();
    }

    private static void sleepNoExp(int seconds) {
        try {
            TimeUnit.SECONDS.sleep(seconds);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    private static int sumOfStr(String str) {
        return Stream.of(str).flatMap(e -> Stream.of(e.split(","))).mapToInt(Integer::valueOf).sum();
    }
}
